package zsy.test.java.util.concurrent;

import lombok.extern.slf4j.Slf4j;
import org.junit.jupiter.api.Test;

import java.util.concurrent.ConcurrentLinkedQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * Semaphore+Redis<List> 实现队列执行任务
 * 1. 消息入列时执行
 * 2. 定时执行
 *
 * @author zhangshuaiyin
 * @date 2022/5/23 10:44
 */
@Slf4j
public class SemaphoreDemo {

    /**
     * 模拟Redis队列
     */
    private final ConcurrentLinkedQueue<String> list = new ConcurrentLinkedQueue<>();

    /**
     * 执行任务的线程池
     */
    private final ThreadPoolExecutor threadPoolExecutor;

    /**
     * 信号量
     */
    private final Semaphore signal = new Semaphore(0);

    /**
     * 定时执行事件
     */
    private static final long pollIntervalSeconds = 30;

    public SemaphoreDemo() {
        // 拒绝策略为默认-AbortPolicy 超过最大线程数会抛异常拒绝新任务，可根据实际业务自行选择
        threadPoolExecutor = new ThreadPoolExecutor(1, 2, 30, TimeUnit.MINUTES, new SynchronousQueue<>(),
                new ThreadFactory() {
                    private final AtomicInteger id = new AtomicInteger(0);

                    @Override
                    public Thread newThread(Runnable r) {
                        Thread thread = new Thread(r);
                        thread.setName("execute-thread-" + id.incrementAndGet());
                        thread.setDaemon(true);
                        return thread;
                    }
                });
        threadPoolExecutor.submit(this::execute);
    }

    /**
     * 消息入列 插入Redis
     * <p>
     * 实例参数为任务参数对象
     */
    public void push(String task) {
        // 入列 
        // redisTemplate.opsForList().leftPush(KEY, task);
        list.add(task);
        signal.release();
    }

    /**
     * 消息出列 从redis中读取
     **/
    private String poll() {
        try {
            // return redisTemplate.opsForList().rightPop(KEY);
            return list.poll();
        } catch (Exception e) {
            System.out.println("读取任务消息失败" + e.getLocalizedMessage());
        }
        return null;
    }

    private void execute() {
        for (String task; (task = poll()) != null; ) {
            log.info("执行任务 {}", task);
        }
        try {
            // 不管是否接收到信号，按照定时任务模式执行
            boolean acquire = signal.tryAcquire(pollIntervalSeconds, TimeUnit.SECONDS);
            threadPoolExecutor.submit(this::execute);
        } catch (InterruptedException ignored) {
        }
    }

    @Test
    public void testSemaphore() throws InterruptedException {
        SemaphoreDemo semaphoreDemo = new SemaphoreDemo();
        for (int i = 0; i < 10; i++) {
            // 每五秒一条消息
            TimeUnit.SECONDS.sleep(5);
            semaphoreDemo.push("第" + i + "次执行");
        }
    }
}
